package bsf.baseservice.businessmq.consumer;

import bsf.base.CallBack;
import bsf.base.TimeSpan;
import bsf.baseservice.businessmq.dal.tb_messagequeue_dal;
import bsf.baseservice.businessmq.db.ConsumerBLL;
import bsf.baseservice.businessmq.model.tb_consumer_partition_model;
import bsf.baseservice.businessmq.model.tb_messagequeue_model;
import bsf.baseservice.businessmq.redis.RedisNetCommandListener;
import bsf.baseservice.businessmq.redis.messagelock.BaseMessageLock;
import bsf.baseservice.businessmq.redis.messagelock.ReceiveMessageLock;
import bsf.baseservice.businessmq.systemruntime.log.DebugHelper;
import bsf.baseservice.businessmq.systemruntime.log.LogHelper;
import bsf.baseservice.businessmq.systemruntime.log.ErrorLogHelper;
import bsf.baseservice.businessmq.systemruntime.*;
import bsf.db.DbConn;
import bsf.system.thread.ThreadPool;
import bsf.util.DateUtil;
import bsf.util.ThreadUtil;
import org.apache.commons.lang3.time.DateUtils;

import java.util.Date;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 消费者内部消息队列（消息缓存）
 
*/
public class ReceiveMessageQuque implements AutoCloseable
{
	private PartitionQueue<MQMessage> Queue = new PartitionQueue<MQMessage>();
	private java.util.HashMap<Integer, Long> lastpullququeiddic = new java.util.HashMap<Integer, Long>(); //上一次拉取的缓存消息的消息maxid
	private ConsumerContext Context;
	private java.util.HashMap<String, String> exsittablenames = new java.util.HashMap<String, String>(); //已经验证存在的表名缓存
	private RedisNetCommandListener redislistener;
	private BaseMessageLock messagelock = new ReceiveMessageLock(new TimeSpan(500));
	private Object tiggerGetMessagesTaskLock = new Object();
	private java.util.HashMap<Integer, Exception> errorpartitions = new java.util.HashMap<Integer, Exception>(); //分区错误信息缓存
	private boolean isClosed = false;

	public ReceiveMessageQuque(ConsumerContext context)
	{
		Context = context;
		Context.Quque=this;
		create();
	}

	private void create()
	{
		try
		{
			isClosed = false;
			DebugHelper.writeLine(Context.getMQPathID(), Context.getMQPath(), "ReceiveMessageQuque-Create", "消费者开始创建内部消息队列");
			init();

			ThreadPool.SystemThreadPool.submit(new Runnable() {
				@Override
				public void run() {
					run();
				}
			});

			DebugHelper.writeLine(Context.getMQPathID(), Context.getMQPath(), "ReceiveMessageQuque-Create", "消费者内部消息队列注册消息循环成功");

			redislistener = new RedisNetCommandListener(ConfigHelper.RedisServer);
			redislistener.Name="消费者内部消息队列";

			redislistener.register(new CallBack.Func2<Object,String,String>() {
				@Override
				public Object invoke(String channel, String message) {
					redisListenerCommand(channel, message);
					return null;
				}
			}, Context.ConsumerProvider.getMQPath(), SystemParamConfig.Redis_Channel_Quque + "." + Context.ConsumerProvider.getMQPath());

			DebugHelper.writeLine(Context.getMQPathID(), Context.getMQPath(), "ReceiveMessageQuque-Create", "消费者内部消息队列注册redis消息监听成功");

			LogHelper.writeLine(Context.getMQPathID(), Context.getMQPath(), "ReceiveMessageQuque-Create", "消费者内部队列注册成功");
		}
		catch (RuntimeException exp)
		{
			ErrorLogHelper.writeLine(Context.getMQPathID(), Context.getMQPath(), "ReceiveMessageQuque-Create", "消费者内部消息队列创建失败", exp);
			throw exp;
		}
	}

	private void init()
	{
		DebugHelper.writeLine(Context.getMQPathID(), Context.getMQPath(), "ReceiveMessageQuque-Init", "消费者内部消息队列Init");
		this.Queue.clear(); //清空消息缓存
		lastpullququeiddic = new java.util.HashMap<Integer, Long>(); //清空消息拉取最后id标记
		exsittablenames = new java.util.HashMap<String, String>(); //已经验证存在的表名缓存
		errorpartitions = new java.util.HashMap<Integer, Exception>(); //分区错误信息缓存

		//初始化最后的mqid
		for (tb_consumer_partition_model partition : Context.ConsumerInfo.ConsumerPartitionModels)
		{
			lastpullququeiddic.put(partition.partitionid, partition.lastmqid);
		}
		if (Context.ConsumerProvider.MaxReceiveMQThread <= 0 || (Context.ConsumerProvider.MaxReceiveMQThread > Context.ConsumerInfo.ConsumerPartitionModels.size()))
		{
			if (Context.ConsumerInfo.ConsumerPartitionModels.size() > 0)
			{
				Context.ConsumerProvider.MaxReceiveMQThread = Context.ConsumerInfo.ConsumerPartitionModels.size();
			}
			else
			{
				Context.ConsumerProvider.MaxReceiveMQThread = 1;
			}
		}
		LogHelper.writeLine(Context.getMQPathID(), Context.getMQPath(), "ReceiveMessageQuque-Init", "消费者内部消息队列Init");
	}


	private void run()
	{
		while (isClosed==false)
		{
			boolean issleep = true;
			try
			{
				issleep = tiggerMessagesTask();
				DebugHelper.writeLine(Context.getMQPathID(), Context.getMQPath(), "ReceiveMessageQuque-Run", "消费者内部消息队列循环一轮完成");
			}
			catch (Exception exp)
			{
				ErrorLogHelper.writeLine(Context.getMQPathID(), Context.getMQPath(), "ReceiveMessageQuque-Run", "消费者内部消息队列循环出错。", exp);
			}
			finally
			{
				if (issleep == true && isClosed==false)
				{
					try{Thread.sleep(SystemParamConfig.Consumer_ReceiveMessageQuque_Every_Sleep_Time * 1000);}catch (Exception e){}
				}
			}
		}
	}

	private void redisListenerCommand(String channel, String msg)
	{

		try
		{
			if (isClosed==false)
			{
				messagelock.lock(new CallBack.Func0() {
					@Override
					public Object invoke() {
						DebugHelper.writeLine(Context.getMQPathID(), Context.getMQPath(), "ReceiveMessageQuque-RedisListenerCommand", "消费者Redis开始执行消息命令");
						tiggerMessagesTask();
						DebugHelper.writeLine(Context.getMQPathID(), Context.getMQPath(), "ReceiveMessageQuque-RedisListenerCommand", "消费者Redis消息命令执行完毕");
						return null;
					}
				});
			}
		}
		catch (Exception exp)
		{
			ErrorLogHelper.writeLine(Context.getMQPathID(), Context.getMQPath(), "ReceiveMessageQuque-RedisListenerCommand", "消费者Redis监听消息命令", exp);
		}

	}

	/** 
	 触发消息队列消息拉取任务
	 
	*/
	private boolean tiggerMessagesTask()
	{
		boolean ifsleep = true; //是否要休眠，除非有大量消息堆积则不休眠
				//若无数据，则检查manage数据库当前时间，若manage数据库时间已改变，且数据库已创建新的日表，则使用新表。
					//serverdate = c.GetServerDate();
				//从各个分区拉消息并缓存
				//获取缓存中最长消息分区队列长度
				//从各个分区缓存推送消息消费
				//队列长度检测及判断是否需要持续拉去数据
		//DebugHelper.TimeWatch(Context.GetMQPathID(), Context.GetMQPath(), "TiggerMessagesTask", () => //有队列溢出异常 - 有部分分区有大量消息堆积的情况 - 仅一个任务在进行扫描
		//{
			synchronized (tiggerGetMessagesTaskLock)
			{
				try(DbConn c = DbConn.createConn(Context.ConsumerProvider.Config.ManageConnectString))
				{
					Context.setManageServerTime(c.getServerDate());
				}
				getMessagesFromPartitions();
				int maxcountofpartition = Queue.maxCountOfPartition();
				SendMessagesToConsume();
				if (maxcountofpartition == SystemParamConfig.Consumer_ReceiveMessageQuque_EVERY_PULL_COUNT)
				{
					ifsleep = false;
				}
				if (maxcountofpartition > SystemParamConfig.Consumer_ReceiveMessageQuque_EVERY_PULL_COUNT)
				{
					throw new BusinessMQException(String.format("最大缓存分区队列长度为:%1$s,超过%2$s,队列过长溢出", maxcountofpartition, SystemParamConfig.Consumer_ReceiveMessageQuque_EVERY_PULL_COUNT));
				}
			}
		return ifsleep;
	}

	private void SendMessagesToConsume()
	{
		//Context.ConsumerInfo.ConsumerPartitionModels.ForEach((partition) =>
		//{
		ThreadUtil.parallelFor(ThreadPool.SystemThreadPool, Context.ConsumerProvider.MaxReceiveMQThread, Context.ConsumerInfo.ConsumerPartitionModels.toArray(), new CallBack.Action1<Object>() {
			@Override
			public void invoke(Object o) {
				tb_consumer_partition_model partition = (tb_consumer_partition_model)o;
				MQMessage m = Queue.dequeue(partition.partitionid);
				while (m != null)
				{
					if (isClosed==true)
					{
						return;
					}
					try
					{
						Context.ActionInfo.InnerAction.invoke(m);
						if (m.getIsMarkFinished() == false)
						{
							throw new BusinessMQException(String.format("当前消息未标记为已消费状态,json:%1$s,id:%2$s", m.Model.message, m.Model.id));
						}
						else
						{
							m = Queue.dequeue(partition.partitionid);
						}
					}
					catch (Exception exp)
					{
						String msg = String.format("MQ消息处理出错,json:%1$s,id:%2$s", m.Model.message, m.Model.id);
						if (!errorpartitions.containsKey(partition.partitionid))
						{
							errorpartitions.put(partition.partitionid, new BusinessMQException(msg, exp));
						}
						ErrorLogHelper.writeLine(Context.getMQPathID(), Context.getMQPath(), "ReceiveMessageQuque-SendMessagesToConsume", msg, exp);
						return;
					}
				}
			}
		});
	}

	private void getMessagesFromPartitions()
	{
		for (tb_consumer_partition_model partition : Context.ConsumerInfo.ConsumerPartitionModels)
		{
			if (isClosed==true)
			{
				break;
			}
			try
			{
				if (errorpartitions.containsKey(partition.partitionid))
				{
					throw new BusinessMQException("分区消息处理出现错误", errorpartitions.get(partition.partitionid));
				}
				Long scanlastmqid = lastpullququeiddic.get(partition.partitionid); //上次扫描的mqid
				MQIDInfo mqidinfo = PartitionRuleHelper.getMQIDInfo(scanlastmqid); //解析上次扫描的mqid信息

				ConsumerBLL consumerbll = new ConsumerBLL(); //解析分区信息
				PartitionIDInfo partionidinfo = PartitionRuleHelper.getPartitionIDInfo(partition.partitionid);
				String datanodeconnectstring = consumerbll.getDataNodeConnectString(SystemParamConfig.Consumer_DataNode_ConnectString_Template, Context.ConsumerInfo.DataNodeModelDic.get(partionidinfo.DataNodePartition)); //获取节点连接

				java.util.ArrayList<tb_messagequeue_model> messages = getMessagesOfPatition(partition, scanlastmqid, mqidinfo, datanodeconnectstring, partionidinfo);

				if (messages.size() > 0)
				{
					consumeMessages(messages, partition, mqidinfo);
				}
				else
				{
					checkIfScanNewTable(partition, mqidinfo, datanodeconnectstring, partionidinfo);
				}


			}
			catch (Exception exp)
			{
				ErrorLogHelper.writeLine(partition.id,"","", String.format("MQ消费者端消息循环出错,clientid:%1$s,partitionid:%2$s", partition.consumerclientid, partition.partitionid), exp);

			}
		}
	}

	private java.util.ArrayList<tb_messagequeue_model> getMessagesOfPatition(tb_consumer_partition_model partition, long scanlastmqid, MQIDInfo mqidinfo, String datanodeconnectstring, PartitionIDInfo partionidinfo)
	{
		//从数据库取数据,检查数据时间，和数据状态，无问题则插入队列，并修改上一次扫描的mqid
		java.util.ArrayList<tb_messagequeue_model> messages = new java.util.ArrayList<tb_messagequeue_model>();
		try(DbConn c = DbConn.createConn(datanodeconnectstring))
		{
			String tablename = PartitionRuleHelper.getTableName(partionidinfo.TablePartition, mqidinfo.Day);
			if (exsittablenames.containsKey(tablename) || c.tableIsExist(tablename))
			{
				addExsitTableNameCache(tablename);
				tb_messagequeue_dal dal = new tb_messagequeue_dal();
				dal.TableName=tablename;
				messages = dal.getMessages(c, scanlastmqid, SystemParamConfig.Consumer_ReceiveMessageQuque_EVERY_PULL_COUNT);
			}
		}
		return messages;
	}

	private void consumeMessages(java.util.ArrayList<tb_messagequeue_model> messages, tb_consumer_partition_model partition, MQIDInfo mqidinfo)
	{
		long maxmqid = -1;
		java.util.ArrayList<MQMessage> mqmessages = new java.util.ArrayList<MQMessage>();
		for (tb_messagequeue_model m : messages)
		{
			//若是第二天的数据或已迁移的数据则跳过
			if ((m.sqlcreatetime.getTime() < DateUtils.addDays(DateUtil.getDate(mqidinfo.Day),1).getTime()) && (m.state == (byte)EnumMessageState.CanRead.Value))
			{
				MQMessage tempVar = new MQMessage();
				tempVar.Model=m;
				tempVar.Context=Context;
				mqmessages.add(tempVar);
			}
			maxmqid = m.id;
		}
		if (mqmessages.size() > 0)
		{
			Queue.enqueue(partition.partitionid, mqmessages);
		}
		if (maxmqid > 0)
		{
			if (lastpullququeiddic.get(partition.partitionid) <= maxmqid && isClosed==false)
			{
				lastpullququeiddic.put(partition.partitionid, maxmqid);
			}
			else
			{
				throw new BusinessMQException(String.format("检测到消费者端拉去消息时出现消息乱序问题,partitionid:%1$s,maxmqid:%2$s,缓存maxmqid:%3$s", partition.partitionid, maxmqid, lastpullququeiddic.get(partition.partitionid)));
			}
		}
	}

	private void checkIfScanNewTable(tb_consumer_partition_model partition, MQIDInfo mqidinfo, String datanodeconnectstring, PartitionIDInfo partionidinfo)
	{
		Date serverdate = DateUtils.addSeconds(Context.getManageServerTime(),1); //此处延迟1秒，停顿消费者日分区表之间的切换
		if (serverdate != null && serverdate.getTime()  > DateUtil.getDate(mqidinfo.Day).getTime()) //服务器时间和消息时间对应不上
		{
			try(DbConn c = DbConn.createConn(datanodeconnectstring))
			{
				Date t = DateUtils.addDays(DateUtil.getDate(mqidinfo.Day),1);
				while (DateUtil.getDate(serverdate).getTime() >= t.getTime())
				{
					String tablename = PartitionRuleHelper.getTableName(partionidinfo.TablePartition, t);
					if (exsittablenames.containsKey(tablename) || c.tableIsExist(tablename))
					{
						addExsitTableNameCache(tablename);
						MQIDInfo tempVar = new MQIDInfo();
						tempVar.AutoID=0;
						tempVar.DataNodePartition=mqidinfo.DataNodePartition;
						tempVar.TablePartition=mqidinfo.TablePartition;
						tempVar.Day=t;
						lastpullququeiddic.put(partition.partitionid, PartitionRuleHelper.getMQID(tempVar));
						break;
					}
					t = DateUtils.addDays(t,1);
				}
			}

		}
	}

	private void addExsitTableNameCache(String tablename)
	{
		if (!exsittablenames.containsKey(tablename))
		{
			exsittablenames.put(tablename, null);
		}
	}

	public final void close()
	{
		if (redislistener != null)
		{
			try{redislistener.close();}catch (Exception exp){}
		}

		if (isClosed==false)
		{
			isClosed = true;
			init();
		}
	}


}